package com.wande.dataplatform.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.wande.common.core.exception.ServiceException;
import com.wande.common.core.utils.StringUtils;
import com.wande.common.mybatis.core.page.PageQuery;
import com.wande.common.mybatis.core.page.TableDataInfo;
import com.wande.dataplatform.common.enums.DataPlatformErrorCode;
import com.wande.dataplatform.domain.CheckRule;
import com.wande.dataplatform.domain.CheckTask;
import com.wande.dataplatform.domain.DataModel;
import com.wande.dataplatform.domain.QualityReport;
import com.wande.dataplatform.domain.bo.CheckTaskBo;
import com.wande.dataplatform.domain.dto.CheckResultDTO;
import com.wande.dataplatform.domain.vo.CheckTaskVo;
import com.wande.dataplatform.mapper.CheckRuleMapper;
import com.wande.dataplatform.mapper.CheckTaskMapper;
import com.wande.dataplatform.mapper.DataModelMapper;
import com.wande.dataplatform.mapper.QualityReportMapper;
import com.wande.dataplatform.service.ICheckTaskService;
import com.wande.dataplatform.service.IDorisService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 数据稽查任务服务实现
 *
 * @author wande
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class CheckTaskServiceImpl implements ICheckTaskService {

    private final CheckTaskMapper checkTaskMapper;
    private final CheckRuleMapper checkRuleMapper;
    private final DataModelMapper dataModelMapper;
    private final QualityReportMapper qualityReportMapper;
    private final IDorisService dorisService;

    @Override
    public TableDataInfo<CheckTaskVo> queryPageList(CheckTaskBo bo, PageQuery pageQuery) {
        LambdaQueryWrapper<CheckTask> wrapper = buildQueryWrapper(bo);
        Page<CheckTask> page = checkTaskMapper.selectPage(pageQuery.build(), wrapper);
        Page<CheckTaskVo> voPage = new Page<>(page.getCurrent(), page.getSize(), page.getTotal());
        List<CheckTaskVo> voList = BeanUtil.copyToList(page.getRecords(), CheckTaskVo.class);
        // 转换ruleIds字段和填充表名
        for (int i = 0; i < page.getRecords().size(); i++) {
            CheckTask entity = page.getRecords().get(i);
            CheckTaskVo vo = voList.get(i);
            // 转换ruleIds
            if (entity.getRuleIds() != null) {
                vo.setRuleIds(JSONUtil.toList(entity.getRuleIds(), Long.class));
            }
            // 填充表名
            if (vo.getTableId() != null) {
                DataModel model = dataModelMapper.selectById(vo.getTableId());
                if (model != null) {
                    vo.setTableName(model.getTableName());
                }
            }
        }
        voPage.setRecords(voList);
        return TableDataInfo.build(voPage);
    }

    @Override
    public List<CheckTaskVo> queryList(CheckTaskBo bo) {
        LambdaQueryWrapper<CheckTask> wrapper = buildQueryWrapper(bo);
        List<CheckTask> list = checkTaskMapper.selectList(wrapper);
        List<CheckTaskVo> voList = BeanUtil.copyToList(list, CheckTaskVo.class);
        // 转换ruleIds字段
        for (int i = 0; i < list.size(); i++) {
            CheckTask entity = list.get(i);
            CheckTaskVo vo = voList.get(i);
            if (entity.getRuleIds() != null) {
                vo.setRuleIds(JSONUtil.toList(entity.getRuleIds(), Long.class));
            }
        }
        return voList;
    }

    @Override
    public CheckTaskVo queryById(Long id) {
        CheckTask entity = checkTaskMapper.selectById(id);
        CheckTaskVo vo = BeanUtil.toBean(entity, CheckTaskVo.class);
        if (vo != null) {
            // 将JSON字符串转换为List<Long>
            if (entity.getRuleIds() != null) {
                vo.setRuleIds(JSONUtil.toList(entity.getRuleIds(), Long.class));
            }
            if (vo.getTableId() != null) {
                DataModel model = dataModelMapper.selectById(vo.getTableId());
                if (model != null) {
                    vo.setTableName(model.getTableName());
                }
            }
        }
        return vo;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Long insertByBo(CheckTaskBo bo) {
        CheckTask entity = BeanUtil.toBean(bo, CheckTask.class);
        // 将List<Long>转换为JSON字符串存储
        entity.setRuleIds(JSONUtil.toJsonStr(bo.getRuleIds()));
        checkTaskMapper.insert(entity);
        return entity.getId();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean updateByBo(CheckTaskBo bo) {
        CheckTask entity = BeanUtil.toBean(bo, CheckTask.class);
        // 将List<Long>转换为JSON字符串存储
        entity.setRuleIds(JSONUtil.toJsonStr(bo.getRuleIds()));
        return checkTaskMapper.updateById(entity) > 0;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean deleteByIds(List<Long> ids) {
        return checkTaskMapper.deleteBatchIds(ids) > 0;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public CheckResultDTO executeCheck(Long taskId) {
        long startTime = System.currentTimeMillis();
        CheckTask task = checkTaskMapper.selectById(taskId);
        if (task == null) {
            throw new ServiceException(DataPlatformErrorCode.ETL_TASK_NOT_FOUND.getMessage());
        }

        // 更新任务状态
        task.setLastExecuteTime(LocalDateTime.now());
        task.setLastExecuteStatus("RUNNING");
        checkTaskMapper.updateById(task);

        CheckResultDTO result = new CheckResultDTO();
        result.setTaskId(taskId);
        result.setTaskName(task.getName());
        result.setTableId(task.getTableId());

        try {
            // 获取数据模型
            DataModel model = dataModelMapper.selectById(task.getTableId());
            if (model == null) {
                throw new ServiceException("数据模型不存在");
            }
            result.setTableName(model.getTableName());

            // 获取规则列表
            List<Long> ruleIds = JSONUtil.toList(task.getRuleIds(), Long.class);
            List<CheckRule> rules = checkRuleMapper.selectBatchIds(ruleIds);

            // 获取表总行数
            String countSql = String.format("SELECT COUNT(*) as total FROM %s", model.getTableName());
            List<Map<String, Object>> countResult = dorisService.executeQuery(countSql);
            Long totalRows = Long.parseLong(countResult.get(0).get("total").toString());
            result.setTotalRows(totalRows);

            // 执行稽查规则
            List<CheckResultDTO.ProblemDetail> problemDetails = new ArrayList<>();
            Map<String, CheckResultDTO.DimensionStat> dimensionStats = new HashMap<>();
            long totalProblemRows = 0;

            for (CheckRule rule : rules) {
                CheckResultDTO.ProblemDetail detail = executeRule(model, rule);
                problemDetails.add(detail);
                totalProblemRows += detail.getProblemRows();

                // 统计维度数据
                String dimension = rule.getDimension();
                CheckResultDTO.DimensionStat stat = dimensionStats.computeIfAbsent(dimension, k -> {
                    CheckResultDTO.DimensionStat s = new CheckResultDTO.DimensionStat();
                    s.setDimension(k);
                    s.setRuleCount(0);
                    s.setProblemRows(0L);
                    return s;
                });
                stat.setRuleCount(stat.getRuleCount() + 1);
                stat.setProblemRows(stat.getProblemRows() + detail.getProblemRows());
            }

            // 计算通过率和质量评分
            for (CheckResultDTO.DimensionStat stat : dimensionStats.values()) {
                BigDecimal passRate = totalRows > 0 ?
                    BigDecimal.valueOf(totalRows - stat.getProblemRows())
                        .divide(BigDecimal.valueOf(totalRows), 4, RoundingMode.HALF_UP)
                        .multiply(BigDecimal.valueOf(100)) : BigDecimal.ZERO;
                stat.setPassRate(passRate);
            }

            result.setProblemRows(totalProblemRows);
            result.setProblemDetails(problemDetails);
            result.setDimensionStats(dimensionStats);

            // 计算质量评分
            BigDecimal qualityScore = totalRows > 0 ?
                BigDecimal.valueOf(totalRows - totalProblemRows)
                    .divide(BigDecimal.valueOf(totalRows), 4, RoundingMode.HALF_UP)
                    .multiply(BigDecimal.valueOf(100)) : BigDecimal.valueOf(100);
            result.setQualityScore(qualityScore);

            result.setStatus("SUCCESS");
            task.setLastExecuteStatus("SUCCESS");

            // 保存质量报告
            saveQualityReport(result, task, model, startTime);

        } catch (Exception e) {
            log.error("执行稽查任务失败", e);
            result.setStatus("FAILED");
            result.setErrorMessage(e.getMessage());
            task.setLastExecuteStatus("FAILED");
        } finally {
            result.setExecuteDuration(System.currentTimeMillis() - startTime);
            checkTaskMapper.updateById(task);
        }

        return result;
    }

    /**
     * 执行单个规则
     */
    private CheckResultDTO.ProblemDetail executeRule(DataModel model, CheckRule rule) {
        CheckResultDTO.ProblemDetail detail = new CheckResultDTO.ProblemDetail();
        detail.setRuleId(rule.getId());
        detail.setRuleName(rule.getName());
        detail.setDimension(rule.getDimension());
        detail.setDescription(rule.getDescription());

        try {
            String checkSql;
            if ("SQL".equals(rule.getRuleType())) {
                // SQL规则直接执行
                checkSql = rule.getRuleContent();
            } else {
                // 其他类型规则暂不支持
                detail.setProblemRows(0L);
                return detail;
            }

            // 执行检查SQL
            List<Map<String, Object>> problemData = dorisService.executeQuery(checkSql);
            detail.setProblemRows((long) problemData.size());

            // 取前10条作为示例
            detail.setSampleData(problemData.stream().limit(10).collect(Collectors.toList()));

        } catch (Exception e) {
            log.error("执行规则失败: {}", rule.getName(), e);
            detail.setProblemRows(0L);
            detail.setDescription("规则执行失败: " + e.getMessage());
        }

        return detail;
    }

    /**
     * 保存质量报告
     */
    private void saveQualityReport(CheckResultDTO result, CheckTask task, DataModel model, long startTime) {
        QualityReport report = new QualityReport();
        report.setReportType("CHECK");
        report.setTaskId(task.getId());
        report.setTaskName(task.getName());
        report.setTableId(model.getId());
        report.setTableName(model.getTableName());
        report.setExecuteTime(LocalDateTime.now());
        report.setTotalRows(result.getTotalRows());
        report.setProblemRows(result.getProblemRows());
        report.setQualityScore(result.getQualityScore());
        report.setExecuteDuration(System.currentTimeMillis() - startTime);
        report.setProblemDetails(JSONUtil.toJsonStr(result.getProblemDetails()));
        report.setDimensionStats(JSONUtil.toJsonStr(result.getDimensionStats()));
        report.setStatus(result.getStatus());
        report.setErrorMessage(result.getErrorMessage());
        report.setProjectId(task.getProjectId());
        qualityReportMapper.insert(report);
    }

    private LambdaQueryWrapper<CheckTask> buildQueryWrapper(CheckTaskBo bo) {
        LambdaQueryWrapper<CheckTask> wrapper = Wrappers.lambdaQuery();
        wrapper.like(StringUtils.isNotBlank(bo.getName()), CheckTask::getName, bo.getName());
        wrapper.eq(bo.getTableId() != null, CheckTask::getTableId, bo.getTableId());
        wrapper.eq(StringUtils.isNotBlank(bo.getScheduleType()), CheckTask::getScheduleType, bo.getScheduleType());
        wrapper.eq(bo.getStatus() != null, CheckTask::getStatus, bo.getStatus());
        wrapper.eq(bo.getProjectId() != null, CheckTask::getProjectId, bo.getProjectId());
        wrapper.orderByDesc(CheckTask::getCreateTime);
        return wrapper;
    }
}
